package com.sijibao.gather.admin.kafka;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.sijibao.gather.admin.TopicAdmin;
@Service("topicAdmin")
public class TopicAdminImpl extends KafkaAdminAdapter implements TopicAdmin
{
	private static final Logger logger = LoggerFactory
			.getLogger(TopicAdminImpl.class);

	public void createTopic(String topic)
	{
		AdminClient adminClient = getAdminClient();
		Collection<NewTopic> newTopics = new ArrayList<NewTopic>();
		String name = topic;
		int numPartitions = 1;
		short replicationFactor = 1;

		NewTopic newTopic = new NewTopic(name, numPartitions,
				replicationFactor);
		newTopics.add(newTopic);

		CreateTopicsResult result=adminClient.createTopics(newTopics);
		KafkaFuture<Void> future=result.all();
		try
		{
			future.get();
		} catch (Exception e)
		{
			logger.error(e.getMessage(),e);
		} 

	}

	public Set<String> queryTopic()
	{
		AdminClient adminClient = getAdminClient();
		ListTopicsResult list = adminClient.listTopics();
		KafkaFuture<Set<String>> nameList = list.names();
		try
		{
			Set<String> nameSet = nameList.get();
			return nameSet;
		} catch (Exception e)
		{
			logger.error(e.getMessage(), e);
			return new HashSet<String>();
		}
	}

}
